package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.AsyncDimFunction;
import com.atguigu.bean.OrderWide;
import com.atguigu.bean.UserProvinceSku;
import com.atguigu.utils.MyClickHouseUtil;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.Tuple3;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class UserProvinceSku10sApp {
    public static void main(String[] args) throws Exception {
        // create runtime environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // ck

        // 从kafka读取数据（upsert-kafka）
        String sourceTopic = "dwd_order_wide";
        String groupId = "user_province_sku_10s_app_2022";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaSource2(sourceTopic, groupId));

        // 将数据转换成javabean并过滤null的数据
        SingleOutputStreamOperator<OrderWide> processDS = kafkaDS.process(new ProcessFunction<String, OrderWide>() {
            @Override
            public void processElement(String value, ProcessFunction<String, OrderWide>.Context ctx, Collector<OrderWide> out) throws Exception {
                try {
                    OrderWide orderWide = JSON.parseObject(value, OrderWide.class);
                    out.collect(orderWide);
                } catch (Exception e) {
                    System.out.println("过滤出upsert-kafka 输出的null值数据");
                }
            }
        });

        processDS.print();

        // upsert-kafka过滤重复数据
        SingleOutputStreamOperator<OrderWide> filterDS = processDS.keyBy(OrderWide::getOrder_detail_id)
                .filter(new RichFilterFunction<OrderWide>() {

                    private ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valuestateDescriptor = new ValueStateDescriptor<>("valuestate", String.class);
                        valuestateDescriptor.enableTimeToLive(new StateTtlConfig.Builder(Time.seconds(5))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build());
                        valueState = getRuntimeContext().getState(valuestateDescriptor);
                    }

                    @Override
                    public boolean filter(OrderWide value) throws Exception {
                        String state = valueState.value();
                        /**
                         * 由于前面的order_info和order_detail的join使用left join 所以需要进行过滤
                         * 如果前面用了join则不需要过滤
                         */
//                        if (state == null && value.getUser_id() != null && value.getProvince_id() != null) {
//                            valueState.update("0");
//                            return true;
//                        }

                        if (state == null) {
                            valueState.update("0");
                            return true;
                        }
                        return false;
                    }
                });

        // 将数据转换成新的javabean
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuDS = filterDS.map(value -> {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

            return UserProvinceSku.builder()
                    .sku_id(value.getSku_id())
                    .sku_name(value.getSku_name())
                    .sku_price(value.getOrder_price())
                    .user_id(value.getUser_id())
                    .province_id(value.getProvince_id())
                    .order_ct(1L)
                    .order_sku_num(value.getSku_num())
                    .order_amount(value.getSplit_total_amount())
                    .ts(sdf.parse(value.getCreate_time()).getTime())
                    .build();

        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<UserProvinceSku>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<UserProvinceSku>() {
                    @Override
                    public long extractTimestamp(UserProvinceSku element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));


        userProvinceSkuDS.print(">>>>>>>>>>>>");

        // 分组开窗聚合
        SingleOutputStreamOperator<UserProvinceSku> result10sDS = userProvinceSkuDS
                .keyBy(new KeySelector<UserProvinceSku, Tuple3<Long, Long, Long>>() {
                    @Override
                    public Tuple3<Long, Long, Long> getKey(UserProvinceSku value) throws Exception {
                        return new Tuple3<>(value.getSku_id(),
                                value.getProvince_id(),
                                value.getUser_id());
                    }
                })
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<UserProvinceSku>() {
                            @Override
                            public UserProvinceSku reduce(UserProvinceSku value1, UserProvinceSku value2) throws Exception {
                                value1.setOrder_ct(value1.getOrder_ct() + value2.getOrder_ct());
                                value1.setOrder_sku_num(value1.getOrder_sku_num() + value2.getOrder_sku_num());
                                value1.setOrder_amount(value1.getOrder_amount().add(value2.getOrder_amount()));
                                return null;
                            }
                        },
                        new WindowFunction<UserProvinceSku, UserProvinceSku, Tuple3<Long, Long, Long>, TimeWindow>() {
                            @Override
                            public void apply(Tuple3<Long, Long, Long> longLongLongTuple3, TimeWindow window, Iterable<UserProvinceSku> input, Collector<UserProvinceSku> out) throws Exception {
                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                                UserProvinceSku next = input.iterator().next();
                                next.setStt(sdf.format(window.getStart()));
                                next.setEdt(sdf.format(window.getEnd()));

                                out.collect(next);
                            }
                        });

        result10sDS.print("reduce10sDS>>>>>>>>>");

        // 关联维度表数据
        // 用户维度表
        SingleOutputStreamOperator<UserProvinceSku> dim_user_info = AsyncDataStream.unorderedWait(result10sDS,
                new AsyncDimFunction<UserProvinceSku>("DIM_USER_INFO") {
                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getUser_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) throws Exception {
//                        input.setUser_age(dimInfo.getLong("GENDER"));
                        SimpleDateFormat sdf = getSimpleDateFormat();
                        input.setUser_gender(dimInfo.getString("GENDER"));
                        String birthday = dimInfo.getString("BIRTHDAY");
                        long age = (System.currentTimeMillis() - sdf.parse(birthday).getTime()) / (1000L * 60 * 60 * 24 * 365);

                        input.setUser_age(age);

                    }
                }, 60, TimeUnit.SECONDS);

        // 地区维度
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithProDS = AsyncDataStream.unorderedWait(dim_user_info,
                new AsyncDimFunction<UserProvinceSku>("DIM_BASE_PROVINCE") {
                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getProvince_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) throws Exception {
                        input.setProvince_name(dimInfo.getString("NAME"));
                        input.setProvince_area_code(dimInfo.getString("AREA_CODE"));
                        input.setProvince_iso_code(dimInfo.getString("ISO_CODE"));
                        input.setProvince_iso_3166_2(dimInfo.getString("ISO_3166_2"));
                    }
                }, 60, TimeUnit.SECONDS);

        // SKU维度表
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithSkuDstream = AsyncDataStream.unorderedWait(userProvinceSkuWithProDS,
                new AsyncDimFunction<UserProvinceSku>("DIM_SKU_INFO") {
                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getSku_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) throws Exception {
                        input.setSpu_id(dimInfo.getLong("SPU_ID"));
                        input.setCategory3_id(dimInfo.getLong("CATEGORY3_ID"));
                        input.setTm_id(dimInfo.getLong("TM_ID"));
                    }
                }, 60, TimeUnit.SECONDS);

        // 关联spu维度表
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithSpuDstream = AsyncDataStream.unorderedWait(userProvinceSkuWithSkuDstream, new AsyncDimFunction<UserProvinceSku>("DIM_SPU_INFO") {
            @Override
            public void join(UserProvinceSku UserProvinceSku, JSONObject jsonObject) throws Exception {
                UserProvinceSku.setSpu_name(jsonObject.getString("SPU_NAME"));
            }

            @Override
            public String getKey(UserProvinceSku UserProvinceSku) {
                return String.valueOf(UserProvinceSku.getSpu_id());
            }
        }, 60, TimeUnit.SECONDS);


        //7.5 关联Category维表 3   2   1
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithCategory3Dstream = AsyncDataStream.unorderedWait(userProvinceSkuWithSpuDstream, new AsyncDimFunction<UserProvinceSku>("DIM_BASE_CATEGORY3") {
            @Override
            public void join(UserProvinceSku UserProvinceSku, JSONObject jsonObject) throws Exception {
                UserProvinceSku.setCategory3_name(jsonObject.getString("NAME"));
                UserProvinceSku.setCategory2_id(jsonObject.getLong("CATEGORY2_ID"));
            }

            @Override
            public String getKey(UserProvinceSku UserProvinceSku) {
                return String.valueOf(UserProvinceSku.getCategory3_id());
            }
        }, 60, TimeUnit.SECONDS);

        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithCategory2Dstream = AsyncDataStream.unorderedWait(userProvinceSkuWithCategory3Dstream, new AsyncDimFunction<UserProvinceSku>("DIM_BASE_CATEGORY2") {
            @Override
            public void join(UserProvinceSku UserProvinceSku, JSONObject jsonObject) throws Exception {
                UserProvinceSku.setCategory2_name(jsonObject.getString("NAME"));
                UserProvinceSku.setCategory1_id(jsonObject.getLong("CATEGORY1_ID"));
            }

            @Override
            public String getKey(UserProvinceSku UserProvinceSku) {
                return String.valueOf(UserProvinceSku.getCategory2_id());
            }
        }, 60, TimeUnit.SECONDS);

        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithCategory1Dstream = AsyncDataStream.unorderedWait(userProvinceSkuWithCategory2Dstream, new AsyncDimFunction<UserProvinceSku>("DIM_BASE_CATEGORY1") {
            @Override
            public void join(UserProvinceSku UserProvinceSku, JSONObject jsonObject) throws Exception {
                UserProvinceSku.setCategory1_name(jsonObject.getString("NAME"));
            }

            @Override
            public String getKey(UserProvinceSku UserProvinceSku) {
                return String.valueOf(UserProvinceSku.getCategory1_id());
            }
        }, 60, TimeUnit.SECONDS);

        //7.6 关联Trademark维表
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithTmDstream = AsyncDataStream.unorderedWait(userProvinceSkuWithCategory1Dstream, new AsyncDimFunction<UserProvinceSku>("DIM_BASE_TRADEMARK") {
            @Override
            public void join(UserProvinceSku UserProvinceSku, JSONObject jsonObject) throws Exception {
                UserProvinceSku.setTm_name(jsonObject.getString("TM_NAME"));
            }

            @Override
            public String getKey(UserProvinceSku UserProvinceSku) {
                return String.valueOf(UserProvinceSku.getTm_id());
            }
        }, 60, TimeUnit.SECONDS);

        userProvinceSkuWithTmDstream.print("userProvinceSkuWithTmDstream>>>>>>>>>>>>>>");

        //TODO 8.将数据写出
        userProvinceSkuWithTmDstream
                .addSink(MyClickHouseUtil.getClickHouseSink("insert into dws_user_province_sku_10s_2022 values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
